package basic.arch.component.lock.distributed.zookeeper;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.GetChildrenBuilder;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCache.Builder;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryNTimes;
import org.apache.curator.utils.CloseableUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.google.common.util.concurrent.ThreadFactoryBuilder;


/**
 * @author:jannal
 * @version V1.0
 */
@Component
public class ZkClientService implements InitializingBean{
    @Autowired
    private ZkConfigInfo zkConfigInfo;

    private static final Logger logger = LoggerFactory.getLogger(ZkClientService.class);

    public static final String ROOT = "/distribution-lock";
    
    public static  CuratorFramework client = null;
    
    
    private  ExecutorService pool = null; 
    
    @Override
    public void afterPropertiesSet() throws Exception {
        init();
    }

    private void initPool(){
        ThreadFactoryBuilder threadFactoryBuilder = new ThreadFactoryBuilder();
        threadFactoryBuilder.setNameFormat("zookeeperListener-pool-%d");
        ThreadFactory threadFactory = threadFactoryBuilder.build();
        pool= Executors.newFixedThreadPool(2,threadFactory); 
    }



    

    private void init() {
        try {
            client = createClient();
            createInitNode();
            initPool();
            pool.submit(new Runnable() {
                
                @Override
                public void run() {
                    try {
                        setListenerTree(ROOT);
                    } catch (Exception e) {
                        logger.error(e.getMessage(),e);
                    }
                }
            });
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            close(client);
            pool.shutdown();
            throw new RuntimeException(e);
        }
    }

    private void setListenerTree(String path) throws Exception {
        Builder newBuilder = TreeCache.newBuilder(client,path);
        TreeCache treeCache = newBuilder.build();
        treeCache.getListenable().addListener(new TreeCacheListener() {

            @Override
            public void childEvent(CuratorFramework client, TreeCacheEvent event) throws Exception {
                ChildData data = event.getData();
                if (data != null) {
                    switch (event.getType()) {
                    case NODE_ADDED:
                        logger.info("添加node路径:{},数据:{}",data.getPath(),new String(data.getData()));
                        break;
                    case NODE_REMOVED:
                        logger.info("移除node路径:{},数据:{}",data.getPath(),new String(data.getData()));
                        break;
                    case NODE_UPDATED:
                        logger.info("修改node路径:{},数据:{}",data.getPath(),new String(data.getData()));
                        break;
                    default:
                        logger.info("其他事件类型{}",event.getType());
                        break;
                    }
                }else{
                    logger.info("数据是null,事件类型是{}",event.getType());
                }
                    
            }
        });
        treeCache.start();
    }

    private CuratorFramework createClient() {
        CuratorFramework client = null;
        try {
            client = CuratorFrameworkFactory.newClient(zkConfigInfo.getHost(),//
                    zkConfigInfo.getSessionTimeoutMs(),//
                    zkConfigInfo.getConnectionTimeoutMs(), //
                    new RetryNTimes(zkConfigInfo.getRetryNTimes(),//
                            zkConfigInfo.getSleepMsBetweenRetries()));

            client.getConnectionStateListenable().addListener(new ConnectionStateListener() {
                @Override
                public void stateChanged(CuratorFramework client, ConnectionState newState) {
                    logger.info("连接状态:{}", newState.name());
                }
            });

            client.start();
        } catch (Exception e) {
            if (client != null) {
                client.close();
            }
            throw new RuntimeException(e);
        }
        return client;
    }

    /**
     * 创建初始化节点
     * 
     * @author jannal
     * @throws Exception
     * @return void
     */
    public CuratorFramework createInitNode() throws Exception {
        String childPath = getInitPath();
        if (!checkExist(childPath)) {
            client.create().withMode(CreateMode.PERSISTENT)// 持久节点
                    .withACL(Ids.OPEN_ACL_UNSAFE)// 所有用户拥有所有权限
                    .forPath(childPath);
            if (logger.isDebugEnabled()) {
                logger.debug("子节点{}创建成功", childPath);
            }
        } else {
            if (logger.isDebugEnabled()) {
                logger.debug("子节点{}已经被创建过", childPath);
            }
        }
        return client;

    }

    private String getInitPath() {
        StringBuilder builder = new StringBuilder();
        builder.append(ROOT);
        String childPath = builder.toString();
        return childPath;
    }

    public void createRoot(String rootPath) throws Exception {
        if (!checkExist(rootPath)) {
            client.create().withMode(CreateMode.PERSISTENT)// 持久节点
                    .withACL(Ids.OPEN_ACL_UNSAFE)// 所有用户拥有所有权限
                    .forPath(ROOT);
            if (logger.isDebugEnabled()) {
                logger.debug("根节点 {}创建成功", rootPath);
            }
        } else {
            if (logger.isDebugEnabled()) {
                logger.debug("根节点{}已经被创建过", rootPath);
            }
        }
    }
    

    public void close(CuratorFramework client) {
        if (client != null) {
            CloseableUtils.closeQuietly(client);
        }
    }

    public boolean checkExist(String path) throws Exception {
        // Stat就是对zonde所有属性的一个映射， stat=null表示节点不存在！
        Stat stat = client.checkExists().forPath(path);
        if (stat == null) {
            return false;
        }
        return true;

    }

    public List<String> getListChildren(String path) throws Exception {
        GetChildrenBuilder children = client.getChildren();
        List<String> paths = children.forPath(path);
        return paths;
    }

   
}
